package com.iotmonitor.iot.emq;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.iotmonitor.iot.domain.dto.*;
import com.iotmonitor.iot.es.ESRepository;
import com.iotmonitor.iot.mapper.TdEngineMapper;
import com.iotmonitor.iot.service.*;
import com.iotmonitor.iot.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.sql.Timestamp;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 接收消息回调类
 */
@Component
@Slf4j
public class EmqMsgProcess implements MqttCallback {

    @Autowired
    private EmqClient emqClient;

    @Autowired
    private ITbQuotaService quotaService;

    @Autowired
    private ITbAlarmService alarmService;

    @Autowired
    private TbDeviceService tbDeviceService;

    @Autowired
    private ITbGpsService gpsService;

    @Autowired
    private ESRepository esRepository;

    @Autowired
    private TbNoticeService noticeService;

    @Autowired
    private TdEngineMapper tdEngineMapper;

    /**
     * 当与MQTT服务器的连接丢失时调用此方法
     * 自动重连
     * @param throwable 它表示导致连接丢失的原因。
     */
    @Override
    public void connectionLost(Throwable throwable) {
        // 连接丢失
        emqClient.connect();
        quotaService.getAllSubject().forEach(s -> {
            // 共享订阅模式
            try {
                emqClient.subscribe("$queue/" + s);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        });
    }


    /**
     * 告警判断和保存设备信息
     * 客户端接收到消息时调用此方法
     *
     * @param topic 参数是主题名称
     * @param mqttMessage 接收到的MqttMessage对象，它包含了接收到的消息内容
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        // 接收到消息
        String payload = new String(mqttMessage.getPayload());
        log.info("接收到数据：{}", payload);
        ObjectMapper mapper = new ObjectMapper();
        Map payloadMap = mapper.readValue(payload, Map.class);

        // 解析指标
        DeviceInfoDTO deviceInfoDTO = quotaService.analysis(topic, payloadMap);
        // TODO查询设备 ，判断开关状态 ，如果是关闭则不处理
        if (deviceInfoDTO != null) {
            // 告警判断
            deviceInfoDTO = alarmService.verifyDeviceInfo(deviceInfoDTO);
            // 保存设备信息
            if (deviceInfoDTO != null){
                boolean b = tbDeviceService.saveDeviceInfo(deviceInfoDTO.getDevice());
                if (b){
                    // 保存指标数据
                    quotaService.saveQuotaToTdengine(deviceInfoDTO.getQuotaList());
                    // 指标透传webHook
                    noticeService.quotaTransfer(deviceInfoDTO.getQuotaList());
                }
            }
        }

        // 解析gps
        DeviceLocation deviceLocation = gpsService.analysis(topic, payloadMap);
        if (deviceLocation != null) {
            log.info("gps解析结果：{}", JsonUtil.serialize(deviceLocation));
            // 保存设备信息
            boolean b = tbDeviceService.saveDeviceInfo(new DeviceDTO(deviceLocation.getDeviceId()));
            if (b){
                // 保存位置信息
                esRepository.saveLocation(deviceLocation);
                // gps透传webHook
                noticeService.gpsTransfer(deviceLocation);
            }
        }

    }

    /**
     * 当消息发送成功并且已经收到服务器的确认时调用此方法
     * @param iMqttDeliveryToken 发送的消息的相关信息，如消息质量等级、主题等
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        try {
            MqttMessage message = iMqttDeliveryToken.getMessage();
            System.out.println(message.getPayload());
            String[] topics = iMqttDeliveryToken.getTopics();
            MqttWireMessage response = iMqttDeliveryToken.getResponse();
            IMqttAsyncClient client = iMqttDeliveryToken.getClient();
            IMqttActionListener actionCallback = iMqttDeliveryToken.getActionCallback();
            int[] grantedQos = iMqttDeliveryToken.getGrantedQos();
            boolean sessionPresent = iMqttDeliveryToken.getSessionPresent();
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }

    }
}
